package com.atguigu.mapreduce.outputformat;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;

public class LogRecordWriter extends RecordWriter<Text, NullWritable> {

    private String atguiguPath = "C:\\major\\SGG大数据线下全套资料2025\\05.尚硅谷大数据技术之Hadoop\\outlog\\atguigu.log";
    private String otherPath = "C:\\major\\SGG大数据线下全套资料2025\\05.尚硅谷大数据技术之Hadoop\\outlog\\other.log";

    private FileSystem fileSystem;
    private FSDataOutputStream atguiguOutputStream;
    private FSDataOutputStream otherOutputStream;

    public LogRecordWriter(TaskAttemptContext context) throws IOException {
        Configuration conf = context.getConfiguration();
        // 获取文件系统对象
        fileSystem = FileSystem.get(conf);
        // 获取输出流对象
        atguiguOutputStream = fileSystem.create(new Path(atguiguPath));
        otherOutputStream = fileSystem.create(new Path(otherPath));
    }

    // 根据业务逻辑将数据写出
    @Override
    public void write(Text key, NullWritable value) throws IOException, InterruptedException {
        // 获取当前写出数据
        String data = key.toString();
        if(data.contains("atguigu")){
            atguiguOutputStream.writeBytes(data + "\n");
        }else {
            otherOutputStream.writeBytes(data + "\n");
        }

    }

    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {
        IOUtils.closeStream(fileSystem);
        IOUtils.closeStream(atguiguOutputStream);
        IOUtils.closeStream(otherOutputStream);
    }
}
